/***********************************************************************
 * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Apache License, Version 2.0
 * which accompanies this distribution and is available at
 * https://www.apache.org/licenses/LICENSE-2.0
 ***********************************************************************/

package org.locationtech.geomesa.redis.data.util

import org.locationtech.geomesa.index.api.BoundedByteRange
import org.locationtech.geomesa.index.utils.AbstractBatchScan
import org.locationtech.geomesa.redis.data.CloseableJedisCommands
import org.locationtech.geomesa.utils.collection.CloseableIterator
import org.locationtech.geomesa.utils.io.WithClose
import redis.clients.jedis.JedisPool
import redis.clients.jedis.util.Pool

private class RedisBatchScan(
    connection: Pool[_ <: CloseableJedisCommands],
    table: Array[Byte],
    ranges: Seq[BoundedByteRange],
    threads: Int,
    buffer: Int
  ) extends AbstractBatchScan[BoundedByteRange, Array[Byte]](ranges, threads, buffer, RedisBatchScan.Sentinel) {

  override protected def scan(range: BoundedByteRange): CloseableIterator[Array[Byte]] = {
    val results = WithClose(connection.getResource)(_.zrangeByLex(table, range.lower, range.upper))
    CloseableIterator(results.iterator())
  }
}

object RedisBatchScan {

  private val Sentinel = new Array[Byte](0)

  def apply(
      connection: Pool[_ <: CloseableJedisCommands],
      table: Array[Byte],
      ranges: Seq[BoundedByteRange],
      threads: Int): CloseableIterator[Array[Byte]] =
    new RedisBatchScan(connection, table, ranges, threads, 100000).start()
}
